package com.kitten.relation_user.biz.consumer;


import com.google.common.util.concurrent.RateLimiter;

import com.kitten.framework.common.utils.DateUtils;
import com.kitten.framework.common.utils.JsonUtils;
import com.kitten.relation_user.biz.constants.MQConstant;
import com.kitten.relation_user.biz.constants.RedisKeyConstant;
import com.kitten.relation_user.biz.domain.dataobject.FansDO;
import com.kitten.relation_user.biz.domain.dataobject.FollowingDO;
import com.kitten.relation_user.biz.domain.mapper.FansDOMapper;
import com.kitten.relation_user.biz.domain.mapper.FollowingDOMapper;
import com.kitten.relation_user.biz.enums.FollowUnFollowTypeEnum;
import com.kitten.relation_user.biz.model.dto.CountFollowUnfollowMqDTO;
import com.kitten.relation_user.biz.model.dto.FollowUserMqDTO;
import com.kitten.relation_user.biz.model.dto.UnfollowUserMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;


import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Objects;


@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_" + MQConstant.TOPIC_FOLLOW_OR_UNFOLLOW,
        consumeMode = ConsumeMode.ORDERLY ,topic = MQConstant.TOPIC_FOLLOW_OR_UNFOLLOW)
public class FollowUnfollowConsumer implements RocketMQListener<Message> {

    @Resource
    private TransactionTemplate transactionTemplate;
    @Autowired
    private FollowingDOMapper followingDOMapper;
    @Autowired
    private FansDOMapper fansDOMapper;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    // Guava 令牌桶, 每秒创建5000个令牌
//    private RateLimiter rateLimiter = RateLimiter.create(5000);
    @Resource
    private RateLimiter rateLimiter;

    /**
     * 处理消息
     * @param message
     */
    @Override
    public void onMessage(Message message) {
        // 流量削峰, 如果没有令牌可用将阻塞, 直到获得
        rateLimiter.acquire();


        String bodyJSONStr = new String(message.getBody());
        // 获取消息Tag
        String tag = message.getTags();
        log.info("=======> User-Relation-Biz Receive message, tag:{}}", tag );

        // 根据 MQ 标签，判断操作类型
        if (Objects.equals(tag, MQConstant.TAG_FOLLOW)) { // 关注
            handleFollowTagMessage(bodyJSONStr);
        } else if (Objects.equals(tag, MQConstant.TAG_UNFOLLOW)) { // 取关
            handleUnFollowTagMessage(bodyJSONStr);
        }
    }

    /**
     * 关注
     * @param bodyJsonStr
     */
    private void handleFollowTagMessage(String bodyJsonStr) {
        FollowUserMqDTO followUserMqDTO = JsonUtils.parseObject(bodyJsonStr, FollowUserMqDTO.class);
        // 判空
        if (Objects.isNull(followUserMqDTO)) {
            return;
        }
        // 联合唯一索引保证数据幂等性 sql: alert table unique
        Long userId = followUserMqDTO.getUserId();
        Long followUserId = followUserMqDTO.getFollowUserId();
        LocalDateTime createTime = followUserMqDTO.getCreateTime();

        // 事务控制
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                // 数据库添加两条记录
                int count = followingDOMapper.insert(FollowingDO.builder()
                        .userId(userId)
                        .followingUserId(followUserId)
                        .createTime(createTime)
                        .build());
                if (count > 0) {
                    fansDOMapper.insert(FansDO.builder()
                            .userId(followUserId)
                            .fansUserId(userId)
                            .createTime(createTime)
                            .build());
                }
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();   // 标记事务为回滚
                log.error("", e);
            }
            return false;
        }));

        log.info("数据库添加关注记录结果: {}", isSuccess);
        // TODO: 更新Redis 的ZSET 粉丝列表(出于业务考虑:关注列表我们不更新)
        // 若数据库操作成功，更新 Redis 中被关注用户的 ZSet 粉丝列表
        if (isSuccess) {
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow_check_and_update_fans_zset.lua")));
            script.setResultType(Long.class);

            // 时间戳
            long timestamp = DateUtils.localDateTime2Timestamp(createTime);
            // 构建被关注用户的粉丝列表的 redis key
            String fansKey = RedisKeyConstant.buildUserFansKey(followUserId);
            // 更新Redis用户粉丝列表, 没有就不更新
            Long result = redisTemplate.execute(script, Collections.singletonList(fansKey), userId, timestamp);
            log.info("更新 Redis 粉丝列表结果: {}", result);

            //TODO: 计数服务
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(followUserId)
                    .type(FollowUnFollowTypeEnum.FOLLOW.getCode())
                    .build();
            // 发送 MQ
            sendMQ(countFollowUnfollowMqDTO);
        }
    }

    /**
     * 取关
     * @param bodyJsonStr
     */
    private void handleUnFollowTagMessage(String bodyJsonStr) {
        UnfollowUserMqDTO unfollowUserMqDTO = JsonUtils.parseObject(bodyJsonStr, UnfollowUserMqDTO.class);
        if (Objects.isNull(unfollowUserMqDTO)) {
            return;
        }
        Long userId = unfollowUserMqDTO.getUserId();
        Long unfollowUserId = unfollowUserMqDTO.getUnfollowUserId();
        LocalDateTime createTime = unfollowUserMqDTO.getCreateTime();

        // 编程式事务
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                int count = followingDOMapper.deleteByUserIdAndFollowingUserId(userId, unfollowUserId); // 关注表删除记录
                if (count > 0) {
                    count = fansDOMapper.deleteByUserIdAndFansUserId(userId, unfollowUserId);   // 粉丝表删除记录
                }
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();   // 标记事务为回滚
                log.error(" ", e);
            }
            return false;
        }));
        // TODO: 若数据库删除成功, 就更新Redis 的ZSET 粉丝列表
        if (isSuccess) {
            String unfollowRedisKey = RedisKeyConstant.buildUserFansKey(unfollowUserId);
            redisTemplate.opsForZSet().remove(unfollowRedisKey, userId);    // 删除粉丝
            // 发送 MQ 通知计数服务：统计关注数
            // 构建消息体 DTO
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(unfollowUserId)
                    .type(FollowUnFollowTypeEnum.UNFOLLOW.getCode()) // 取关
                    .build();

            // 发送 MQ
            sendMQ(countFollowUnfollowMqDTO);
        }
    }







    /**
     * 发送 MQ 通知计数服务
     * @param countFollowUnfollowMqDTO
     */
    private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) {
        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowMqDTO))
                .build();

        // 异步发送 MQ 消息
        // 发送 MQ 通知计数服务：统计关注数
        rocketMQTemplate.asyncSend(MQConstant.TOPIC_COUNT_FOLLOWING, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：关注数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：关注数】MQ 发送异常: ", throwable);
            }
        });

        // 发送 MQ 通知计数服务：统计粉丝数
        rocketMQTemplate.asyncSend(MQConstant.TOPIC_COUNT_FANS, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数】MQ 发送异常: ", throwable);
            }
        });
    }

}